package com.zyx.flinkdemo.stream.window.trigger;

import com.zyx.flinkdemo.pojo.WordCountTs;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeoutTrigger;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author Yaxi.Zhang
 * @since 2021/6/16 11:13
 * desc: ContinuousEventTimeTrigger与ProcessingTimeoutTrigger联合使用案例
 *  实现在指定窗口内每隔一段时间触发一次计算, 同时在数据超时的时候也会触发计算
 */
public class ContinuEvTimeWithProcTimeout {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WordCountTs> wordCountDs = env
                .socketTextStream("localhost", 7777)
                .map(new MapFunction<String, WordCountTs>() {
                    @Override
                    public WordCountTs map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WordCountTs(split[0],  Integer.parseInt(split[1]), split[2], Long.parseLong(split[3]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WordCountTs>forMonotonousTimestamps()
                        .withTimestampAssigner((SerializableTimestampAssigner<WordCountTs>) (element, recordTimestamp) -> element.getTs()));


        wordCountDs
                .keyBy(WordCountTs::getWord)
                // 开间隔为20s的滚动窗口
                .window(TumblingEventTimeWindows.of(Time.of(20, TimeUnit.SECONDS)))
                /*
                  使用
                        ProcessingTimeoutTrigger作为触发器, 超时时间为5s
                        内部的触发器使用ContinuousEventTimeTrigger, 每5s(事件时间)触发一次计算
                  效果
                        在事件时间内, 以20s为窗口, 窗口内每5s更新一次数据,
                        如果期间有5s(处理时间)都没有数据, 则会触发超时操作进行计算
                 */
                .trigger(ProcessingTimeoutTrigger.of(ContinuousEventTimeTrigger.of(Time.seconds(5)), Duration.ofSeconds(5L)))
                .reduce((ReduceFunction<WordCountTs>) (value1, value2) -> new WordCountTs(value1.getWord(),
                        value1.getCount() + value2.getCount(),
                        value1.getTime().compareTo(value2.getTime()) >= 0 ? value1.getTime() : value2.getTime(),
                        Math.max(value1.getTs(), value2.getTs())))
                .print();

        env.execute();
    }
}
